package com.innodealing.consumer;

import com.innodealing.offsetskip.annotation.KafkaOffsetSkip;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

/**
 * 订单处理消费者
 * 演示 groupId 通过配置文件指定，而不是在 @KafkaListener 注解中硬编码
 */
@Service
public class SingleAutoConsumer {
    @KafkaListener(
            topics = "single-auto-example",
            containerFactory = "singleAutoKafkaListenerContainerFactory"
    )
    @KafkaOffsetSkip
    public void handleOrderCreated(ConsumerRecord<String, String> record) {
        String value = record.value();

        try {
            // 模拟业务处理逻辑
            if (value.contains("error")) {
                throw new RuntimeException("[single-auto-example] 业务处理异常: " + value);
            }

            System.out.println("[single-auto-example] 处理成功: " + value);
        } catch (Exception e) {
            System.err.println("分区号: " + record.partition() + "offset: " + record.offset() + "，消息处理失败，1秒后将自动重试: " + value + ", 错误: " + e.getMessage());
            throw e;
        }
    }

}
